package com.ylkj.bus.device.report;

import com.alibaba.fastjson2.JSONObject;
import com.ylkj.dao.RdsUtils;

import com.ylkj.stream.rds.RdsConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.text.MessageFormat;
import java.time.Duration;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @Author: wengzx
 * @Date: 2025/4/18
 * @Desc:
 */
@Slf4j
public class UpdateDevicReportAndUnpack extends ProcessFunction<String, String> {
    // 基础插入SQL模板
    private static final String INSERT_SQL_TEMPLATE =
            "INSERT INTO {0} (`id`, `device_code`,  `report_time`, {1}, `operator_id`, `create_time`, `update_time`) " +
                    "VALUES (?, ?, ?, ?, ?, ? , ?) " +
                    "ON DUPLICATE KEY UPDATE " +
                    "`device_code` = VALUES(`device_code`), " +
                    "`report_time` = VALUES(`report_time`), " +
                    "`{1}` = VALUES(`{1}`), " +  // 动态字段
                    "`operator_id` = VALUES(`operator_id`), " +
                    "`create_time` = VALUES(`create_time`), " +
                    "`update_time` = VALUES(`update_time`)";
    // 基础删除SQL模板
    private static final String DELETE_SQL_TEMPLATE = "DELETE FROM {0} WHERE report_time < ?";

    // 定时任务相关
    private ScheduledExecutorService scheduler;

    // 数据库配置和表名
    private final String tableName;
    private final RdsConfig rdsConfig;
    private RdsUtils rdsUtils;

    // SQL缓存（提升性能）
    private final Map<String, String> insertSqlCache = new ConcurrentHashMap<>();

    // 构造函数，用于初始化
    public UpdateDevicReportAndUnpack(String tableName, RdsConfig rdsConfig) {
        this.rdsConfig = rdsConfig;
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化数据库工具类
        rdsUtils = new RdsUtils(rdsConfig);
        // 创建定时任务调度器
        scheduler = Executors.newSingleThreadScheduledExecutor();
        // 立即执行一次清理任务
        deleteOldData();
        long initialDelay = calculateInitialDelay();
        scheduler.scheduleAtFixedRate(
                this::deleteOldData,
                initialDelay,
                TimeUnit.DAYS.toSeconds(1),
                TimeUnit.SECONDS
        );
        log.info("定时清理任务已启动，首次执行时间：{}秒后", initialDelay / 1000);
    }

    @Override
    public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
        if (StringUtils.isBlank(value)) {
            log.warn("收到空值数据，跳过处理");
            return;
        }
        try {
            JSONObject jsonObject = JSONObject.parseObject(value);
            String deviceCode = jsonObject.getString("deviceCode");
            String clientId = jsonObject.getString("clientId");
            String onlineStatus = jsonObject.getString("onlineStatus");
            String reportTime = jsonObject.getString("reportTime");
            String lastDownTime = jsonObject.getString("lastDownTime");
            String lastUpTime = jsonObject.getString("lastUpTime");

            // 时间格式校验
            DateTimeFormatter originalFormat = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            LocalDateTime dateTime;
            try {
                dateTime = LocalDateTime.parse(reportTime, originalFormat);
            } catch (DateTimeParseException e) {
                log.error("时间格式错误，预期格式：yyyy-MM-dd HH:mm:ss，实际值：{}", reportTime, e);
                return;
            }

            // 生成唯一ID（设备码_日期_类型）
            String id = String.format(
                    "%s_%s",
                    deviceCode,
                    dateTime.format(DateTimeFormatter.ofPattern("yyyyMMdd"))
            );
            // 生成动态字段名（如 online_status__0000）
            String hhmm = reportTime.substring(11, 16).replace(":", "");
            String fieldName = "online_status_" + hhmm;
            // 获取/缓存SQL语句
            String insertSql = insertSqlCache.computeIfAbsent(
                    fieldName,
                    k -> MessageFormat.format(INSERT_SQL_TEMPLATE, tableName, k)
            );

            // 数据库操作
            try (Connection connection = rdsUtils.getConnection()) {
                connection.setAutoCommit(false);
                try (PreparedStatement stmt = connection.prepareStatement(insertSql)) {
                    stmt.setString(1, id);
                    stmt.setString(2, deviceCode);
                    stmt.setString(3, dateTime.format(DateTimeFormatter.ofPattern("yyyyMMdd"))); // 修正：直接使用原始时间格式
                    stmt.setString(4, onlineStatus);
                    stmt.setString(5, ""); // operator_id 默认空
                    stmt.setString(6, reportTime);
                    stmt.setString(7, reportTime);
                    stmt.executeUpdate();
                }

                connection.commit();
                log.debug("成功提交设备数据，设备码：{}，时间：{}", deviceCode, reportTime); // 替换为实际处理的数据信息
            }
        } catch (Exception e) {
            log.error("数据处理失败: {}", e.getMessage(), e);
            throw e; // 抛出异常触发Flink重试机制
        }
    }

    /**
     * 定时清理三天前的旧数据
     */
    private void deleteOldData() {
        try (Connection connection = rdsUtils.getConnection();
             PreparedStatement stmt = connection.prepareStatement(
                     MessageFormat.format(DELETE_SQL_TEMPLATE, tableName))) {

            // 计算清理时间（当前时间三天前）
            LocalDate threeDaysAgo = LocalDate.now().minusDays(3);
            stmt.setDate(1, java.sql.Date.valueOf(threeDaysAgo));

            int rowsDeleted = stmt.executeUpdate();
            log.info("已清理{}条超过3天的旧数据", rowsDeleted);
        } catch (Exception e) {
            log.error("数据清理失败: {}", e.getMessage(), e);
        }
    }

    private long calculateInitialDelay() {
        LocalDateTime nextRun = LocalDateTime.of(
                LocalDateTime.now().toLocalDate(),
                LocalTime.MIDNIGHT
        );
        if (LocalDateTime.now().isAfter(nextRun)) {
            nextRun = nextRun.plusDays(1);
        }
        return Duration.between(LocalDateTime.now(), nextRun).toMillis();
    }

    @Override
    public void close() throws Exception {
        super.close();
        rdsUtils.close();
        scheduler.shutdown();
        log.info("定时清理任务已关闭");
    }
}
